package com.iflyun.LogsCustmerToHDFS.utils;


import java.io.BufferedInputStream;
import java.io.ByteArrayInputStream;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;


import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Progressable;


public class Test {
	

	public static void main(String[] args) throws Exception {
		  try {
//		   uploadToHdfs();   
//		   deleteFromHdfs();
		   getDirectoryFromHdfs();
//		   appendToHdfs();
//		   readFromHdfs();
//		   deleteDir("log/web") ;
		  } catch (Exception e) {
		   e.printStackTrace();
		  }
		  finally
		  {
		   System.out.println("SUCCESS");
		  }
		 }

		 

		 /**上传文件到HDFS上去*/

		 private static void uploadToHdfs() throws FileNotFoundException,IOException {
		  String localSrc = "d://qq.txt";
		  String dst = "hdfs://172.16.10.235:8022/log/api/2017-08-14/qq.txt";
		  InputStream in = new BufferedInputStream(new FileInputStream(localSrc));
		  Configuration conf = new Configuration();
		  
		  FileSystem fs = FileSystem.get(URI.create(dst), conf);
		  OutputStream out = fs.create(new Path(dst), new Progressable() {
		   public void progress() {
		    System.out.print(".");
		   }
		  });
		  IOUtils.copyBytes(in, out, 4096, true);
		 }
		 

		 /**从HDFS上读取文件*/
		 private static void readFromHdfs() throws FileNotFoundException,IOException {
		  String dst = "hdfs://172.16.10.235:8022/log/api/2017-08-23.txt";  
		  Configuration conf = new Configuration();  
		  FileSystem fs = FileSystem.get(URI.create(dst), conf);
		  FSDataInputStream hdfsInStream = fs.open(new Path(dst));
		  
		  OutputStream out = new FileOutputStream("d:/qq-hdfs.txt"); 
		  byte[] ioBuffer = new byte[1024];
		  int readLen = hdfsInStream.read(ioBuffer);

		  while(-1 != readLen){
		  out.write(ioBuffer, 0, readLen);  
		  readLen = hdfsInStream.read(ioBuffer);
		  }
		  out.close();
		  hdfsInStream.close();
		  fs.close();
		 }
		 

		 /**以append方式将内容添加到HDFS上文件的末尾;注意：文件更新，需要在hdfs-site.xml中添<property><name>dfs.append.support</name><value>true</value></property>*/
		 private static void appendToHdfs() throws FileNotFoundException,IOException {
		  
		  String hdfs_path = "hdfs://172.16.10.235:8022/log/api/2017-08-14/2017-08-14.txt";//文件路径
	      Configuration conf = new Configuration();
	      conf.setBoolean("dfs.client.block.write.replace-datanode-on-failure.enable", true);
	      conf.set("dfs.client.block.write.replace-datanode-on-failure.policy", "NEVER");
	     
	        FileSystem fs = null;
	        try {
	            fs = FileSystem.get(URI.create(hdfs_path), conf);
	            InputStream in = new ByteArrayInputStream("哈哈哈123124123123".getBytes("UTF-8"));
	            boolean flag= fs.exists(new Path(hdfs_path));
	            OutputStream out=null;
	            if(flag) {
	            	out =fs.append(new Path(hdfs_path));
	            }else {
	            	out = fs.create(new Path(hdfs_path));
	            }
	            IOUtils.copyBytes(in, out, 4096, true);
	        } catch (IOException e) {
	            e.printStackTrace();
	        }
		 }
		 

		 /**从HDFS上删除文件*/
		 private static void deleteFromHdfs() throws FileNotFoundException,IOException {
		  String dst = "hdfs://172.16.10.85:9000/user/hadoop/qq-bak.txt";  
		  Configuration conf = new Configuration();  
		  FileSystem fs = FileSystem.get(URI.create(dst), conf);
		  fs.deleteOnExit(new Path(dst));
		  fs.close();
		 }
		 

		 /**遍历HDFS上的文件和目录*/
		 private static void getDirectoryFromHdfs() throws FileNotFoundException,IOException {
		  System.setProperty("HADOOP_USER_NAME","root");
		  System.setProperty("HADOOP_USER_PASSWORD","xflc@2015");
		  String dst = "hdfs://172.16.10.235:8022/log/api";  
		  Configuration conf = new Configuration();  
		  FileSystem fs = FileSystem.get(URI.create(dst), conf);
		  FileStatus fileList[] = fs.listStatus(new Path(dst));
		  int size = fileList.length;
		  for(int i = 0; i < size; i++){
			  System.out.println("name:" + fileList[i].getPath().getName() + "\tsize:" + fileList[i].getLen());
		  }
		  fs.close();
		 } 
		 
		 
		 public static boolean deleteDir(String dir) throws IOException {  
		        if (StringUtils.isBlank(dir)) {  
		            return false;  
		        }  
		        dir = "hdfs://172.16.10.235:8022/" + dir;  
		        Configuration conf = new Configuration();  
		        FileSystem fs = FileSystem.get(URI.create(dir), conf);  
		        fs.delete(new Path(dir), true);  
		        fs.close();  
		        return true;  
		    }  

}
